package sf.config.reactive;

import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.transaction.support.TransactionTemplate;
import reactor.core.scheduler.Scheduler;
import reactor.core.scheduler.Schedulers;

import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;

@Configuration
public class SchedulerConfiguration {

    @Value("${spring.datasource.maximum-pool-size:200}")
    private Integer maxPoolSize;
    private static Integer CORE_POOL_SIZE;

    public static final String EXECUTE_SERVICE_NAME = "esPool";

    static {
        CORE_POOL_SIZE = Runtime.getRuntime().availableProcessors();
        if (CORE_POOL_SIZE < 0) {
            CORE_POOL_SIZE = 4;
        }
    }

    @Bean(name = EXECUTE_SERVICE_NAME)
    public ExecutorService esPool() {
        if (maxPoolSize < CORE_POOL_SIZE) {
            maxPoolSize = CORE_POOL_SIZE;
        }
//        return Schedulers.fromExecutor(Executors.newFixedThreadPool(connectionPoolSize));
        ThreadPoolExecutor pool = new ThreadPoolExecutor(maxPoolSize, maxPoolSize, 1L,
                TimeUnit.MINUTES, new ArrayBlockingQueue<>(maxPoolSize * 2), // 使用有界队列，避免OOM
                new ThreadPoolExecutor.CallerRunsPolicy());
        pool.allowCoreThreadTimeOut(true);//允许回收核心线程.
        return pool;
    }

    @Bean
    public Scheduler scheduler(@Autowired @Qualifier(EXECUTE_SERVICE_NAME) ExecutorService esPool) {
        return Schedulers.fromExecutor(esPool);
    }

    @Bean
    public ReactiveTransactionTemplate reactiveTransactionTemplate(TransactionTemplate tt, Scheduler scheduler) {
        return new ReactiveTransactionTemplate(tt, scheduler);
    }
}